<?
/**
 * 本文件用于与Cassandra服务器交互
 *
 * @author    徐鹏程 <xupengcheng@snda.com>
 */


if (!isset($GLOBALS['THRIFT_ROOT'])) $GLOBALS['THRIFT_ROOT'] =  dirname(__FILE__)."/thrift";
require_once $GLOBALS['THRIFT_ROOT'].'/packages/cassandra/Cassandra.php';
require_once $GLOBALS['THRIFT_ROOT'].'/packages/cassandra/cassandra_types.php';
require_once $GLOBALS['THRIFT_ROOT'].'/transport/TSocket.php';
require_once $GLOBALS['THRIFT_ROOT'].'/protocol/TBinaryProtocol.php';
require_once $GLOBALS['THRIFT_ROOT'].'/transport/TFramedTransport.php';
require_once $GLOBALS['THRIFT_ROOT'].'/transport/TBufferedTransport.php';


require_once(dirname(__FILE__).'/../config/UCConfCassandra.php');

if (defined('__UC_ENV') && __UC_ENV == 'DEV') {
} else if (defined('__UC_ENV') && __UC_ENV == 'TEST') {
	if (!defined("DISABLE_CASSANDRA_WRITE")) define("DISABLE_CASSANDRA_WRITE", 1);
	if (!defined("DISABLE_CASSANDRA_READ")) define("DISABLE_CASSANDRA_READ", 1);
} else {
	if (!defined("DISABLE_CASSANDRA_WRITE")) define("DISABLE_CASSANDRA_WRITE", 1);
	if (!defined("DISABLE_CASSANDRA_READ")) define("DISABLE_CASSANDRA_READ", 1);
}

class UCLibCassandra {

	static $error;

	public static function add($mutation_map, $consistency_level = cassandra_ConsistencyLevel::ZERO, $keyspace = "UC") {
		$_time_start = microtime(true);
		$time = microtime(true);
		$mm = array();
		foreach ($mutation_map as $key => &$cfs) {
			$mm[$key] = array();
			foreach ($cfs as $cf=>$data) {
				$mm[$key][$cf]= self::array_to_mutations($data, $time);
			}
		}

		// Make a connection to the Thrift interface to Cassandra
		$servers = UCConfCassandra::getServers();
		$cnt = count($servers);
		$max = min(2, $cnt);

		$index = mt_rand(0, $cnt - 1);
		$i = 0;
		$host = null;
		$port = null;
		while ($i < $max) {
			try {
				$host = $servers[$index][0];
				$port = $servers[$index][1];
				$socket = new TSocket($host, $port);
				$transport = new TBufferedTransport($socket, 1024, 1024);
				if (function_exists('thrift_protocol_write_binary')) {
					$protocol = new TBinaryProtocolAccelerated($transport);
				} else {
					$protocol = new TBinaryProtocol($transport);
				}
				$client = new CassandraClient($protocol);
				$transport->open();
				$client->batch_mutate($keyspace, $mm, $consistency_level);
				$transport->close();
//				UCLibLogger::rpc("add2Q:".$channel.':'.$user_id.":".$record_id, microtime(true) - $_time_start, $host, $port, UCLIB_LOG_RPC_TYPE_MIS);
				return true;
			} catch (Exception $e) {
				self::$error = 'TException: '.$e->why. ' Error: '.$e->getMessage() . "\n";
			}
			$i += 1;
			$index = ($index + 1) % $cnt;
		}
//		UCLibLogger::rpc("check:".$channel.':'.$user_id.":".$record_id, microtime(true) - $_time_start, '', '', UCLIB_LOG_RPC_TYPE_MIS);
		return false;
	}

	public static function get($key, $column_family, $columns, $consistency_level = cassandra_ConsistencyLevel::ONE, $keyspace = "UC") {
		$_time_start = microtime(true);
		
		$cp = new cassandra_ColumnPath();
		$cp->column_family = $column_family;

		$sp = new cassandra_SlicePredicate();
		$sp->column_names = $columns;

		// Make a connection to the Thrift interface to Cassandra
		$servers = UCConfCassandra::getServers();
		$cnt = count($servers);
		$max = min(2, $cnt);

		$index = mt_rand(0, $cnt - 1);
		$i = 0;
		$host = null;
		$port = null;
		while ($i < $max) {
			try {
				$host = $servers[$index][0];
				$port = $servers[$index][1];
				$socket = new TSocket($host, $port);
				$transport = new TBufferedTransport($socket, 1024, 1024);
				if (function_exists('thrift_protocol_write_binary')) {
					$protocol = new TBinaryProtocolAccelerated($transport);
				} else {
					$protocol = new TBinaryProtocol($transport);
				}
				$client = new CassandraClient($protocol);
				$transport->open();
				$slices = $client->get_slice($keyspace, $key, $cp, $sp, $consistency_level);
				$transport->close();
//				UCLibLogger::rpc("add2Q:".$channel.':'.$user_id.":".$record_id, microtime(true) - $_time_start, $host, $port, UCLIB_LOG_RPC_TYPE_MIS);
				$ret = array();
				foreach($slices as $slice) {
					$arr = self::column_or_supercolumn_to_array($slice);
					foreach($arr as $k => $v)
						$ret[$k] = $v;
				}
				return $ret;
			} catch (cassandra_NotFoundException $e) {
				self::$error = 'TException: '.$e->why. ' Error: '.$e->getMessage() . "\n";
				return null;
			} catch (Exception $e) {
				self::$error = 'TException: '.$e->why. ' Error: '.$e->getMessage() . "\n";
			}
			$i += 1;
			$index = ($index + 1) % $cnt;
		}
//		UCLibLogger::rpc("check:".$channel.':'.$user_id.":".$record_id, microtime(true) - $_time_start, '', '', UCLIB_LOG_RPC_TYPE_MIS);
		return false;
	}


	public static function get_slice($key, $path, $start, $count, $reversed=0, $consistency_level = cassandra_ConsistencyLevel::ONE, $keyspace = "UC") {
		$_time_start = microtime(true);
		
		$cp = self::array_to_column_parent($path);

		$sp = new cassandra_SlicePredicate();
		//$sp->column_names = array("b16bde0064a511df8b1e000c29d0ebbd");
		$sp->slice_range = new cassandra_SliceRange();
		$sp->slice_range->start = $start;
		$sp->slice_range->reversed = $reversed;
		$sp->slice_range->finish = "";
		$sp->slice_range->count = $count;


		// Make a connection to the Thrift interface to Cassandra
		$servers = UCConfCassandra::getServers();
		$cnt = count($servers);
		$max = min(2, $cnt);

		$index = mt_rand(0, $cnt - 1);
		$i = 0;
		$host = null;
		$port = null;
		while ($i < $max) {
			try {
				$host = $servers[$index][0];
				$port = $servers[$index][1];
				$socket = new TSocket($host, $port);
				$transport = new TBufferedTransport($socket, 1024, 1024);
				if (function_exists('thrift_protocol_write_binary')) {
					$protocol = new TBinaryProtocolAccelerated($transport);
				} else {
					$protocol = new TBinaryProtocol($transport);
				}
				$client = new CassandraClient($protocol);
				$transport->open();
				$ret = $client->get_slice($keyspace, $key, $cp,  $sp, $consistency_level);
				$transport->close();
//				UCLibLogger::rpc("add2Q:".$channel.':'.$user_id.":".$record_id, microtime(true) - $_time_start, $host, $port, UCLIB_LOG_RPC_TYPE_MIS);
				$array = array();
				if (is_array($ret)) {
					foreach ($ret as $col) {
						$array []= self::column_or_supercolumn_to_array($col);
					}
				}
				return $array;
			} catch (Exception $e) {
				self::$error = 'TException: '.$e->why. ' Error: '.$e->getMessage() . "\n";
			}
			$i += 1;
			$index = ($index + 1) % $cnt;
		}
//		UCLibLogger::rpc("check:".$channel.':'.$user_id.":".$record_id, microtime(true) - $_time_start, '', '', UCLIB_LOG_RPC_TYPE_MIS);
		return false;
	}

	public static function count($key, $path, $consistency_level = cassandra_ConsistencyLevel::ONE, $keyspace = "UC") {
		$_time_start = microtime(true);
		
		$cp = self::array_to_column_parent($path);
		// Make a connection to the Thrift interface to Cassandra
		$servers = UCConfCassandra::getServers();
		$cnt = count($servers);
		$max = min(2, $cnt);

		$index = mt_rand(0, $cnt - 1);
		$i = 0;
		$host = null;
		$port = null;
		while ($i < $max) {
			try {
				$host = $servers[$index][0];
				$port = $servers[$index][1];
				$socket = new TSocket($host, $port);
				$transport = new TBufferedTransport($socket, 1024, 1024);
				if (function_exists('thrift_protocol_write_binary')) {
					$protocol = new TBinaryProtocolAccelerated($transport);
				} else {
					$protocol = new TBinaryProtocol($transport);
				}
				$client = new CassandraClient($protocol);
				$transport->open();
				$ret = $client->get_count($keyspace, $key, $cp, $consistency_level);
				$transport->close();
//				UCLibLogger::rpc("add2Q:".$channel.':'.$user_id.":".$record_id, microtime(true) - $_time_start, $host, $port, UCLIB_LOG_RPC_TYPE_MIS);
				return $ret;
			} catch (Exception $e) {
				self::$error = 'TException: '.$e->why. ' Error: '.$e->getMessage() . "\n";
			}
			$i += 1;
			$index = ($index + 1) % $cnt;
		}
//		UCLibLogger::rpc("check:".$channel.':'.$user_id.":".$record_id, microtime(true) - $_time_start, '', '', UCLIB_LOG_RPC_TYPE_MIS);
		return false;
	}

	public static function test() {
/*		$mutation_map = array();
		$mutation_map["abcd"] = array(
			"twitter" => array (
				"title" => "this title",
				"content" => "this is content"
			)
		);

		$mutation_map["abcd"] = array(
			"retwitter" => array (
				"124" => array(
					"content" => "this is content"
				)
			)
		);
		var_dump(self::add($mutation_map));*/
		#var_dump(self::get_slice("1212175243","twitter", "a5dd7c6c64a311df8b1e000c29d0ebbd", 10, 1));
		#var_dump(self::$error);
		var_dump(self::get("4bf753ed019522e03dcfe267039c2998", "twitter", array("content")));
		var_dump(self::$error);
		//var_dump(self::count("1236c5b8649911df8b1e000c29d0ebbd","retwitter"));
		//var_dump(self::$error);
	}

	static function format($str, $len) {
		$l = strlen($str);
		if ($l == $len) return $str;
		if ($l > $len) return substr($str, $l - $len);
		return str_repeat('0', $len - $l) . $str;
	}

	static function uuid32() {
		$ret = "";
		$time = microtime(true);

		// 时间戳16位
		$ret .= self::format(dechex(intval($time)), 8);
		$ret .= self::format(dechex(($time - intval($time)) * 1000000000), 8);
		// 随机值16位
		$ret .= self::format(dechex(mt_rand()), 8);
		$ret .= self::format(dechex(mt_rand()), 8);

		return $ret;
	}

	static function uuid16() {
		$ret = "";
		$time = microtime(true);

		// 时间戳12位
		$ret .= self::format(dechex(intval($time)), 8);
		$ret .= self::format(dechex(($time - intval($time)) * 1000000), 4);
		// 随机值4位
		$ret .= format(dechex(mt_rand()), 4);

		return $ret;
	}


	static function array_to_mutations($array, $timestamp) {
		$ret = array();
		foreach($array as $name => $value) {
			$c_or_sc = new cassandra_ColumnOrSuperColumn();
			if (is_array($value)) {
				$c_or_sc->super_column = new cassandra_SuperColumn();
				$c_or_sc->super_column->name = $name;
				$c_or_sc->super_column->columns = array();

				foreach($value as $k => $v) { 
					$column = new cassandra_Column();
					$column->name = $k;
					$column->value = $v;
					$column->timestamp = $timestamp;
					$c_or_sc->super_column->columns []= $column;
				}
				$c_or_sc->super_column->timestamp = $timestamp;
			} else {
				$c_or_sc = new cassandra_ColumnOrSuperColumn();
				$c_or_sc->column = new cassandra_Column();
				$c_or_sc->column->name = $name;
				$c_or_sc->column->value = $value;
				$c_or_sc->column->timestamp = $timestamp;
			}
			$ret[] = new cassandra_Mutation(array('column_or_supercolumn' => $c_or_sc));
		}
		return $ret;
	}

	static function array_to_column_path($array) {
		if (is_array($array)) {
			$cp = new cassandra_ColumnPath();
			$cp->column_family = $array[0];
			if (count($array) > 1)
				$cp->super_column = $array[1];
		} else {
			$cp = new cassandra_ColumnPath();
			$cp->column_family = $array;
		}
		return $cp;
	}

	static function array_to_column_parent($array) {
		if (is_array($array)) {
			if (count($array) < 1) return null;
			$cp = new cassandra_ColumnParent();
			$cp->column_family = $array[0];
			if (count($array) > 1)
				$cp->super_column = $array[1];
		} else {
			$cp = new cassandra_ColumnParent();
			$cp->column_family = $array;
		}
		return $cp;
	}

	
	static function column_or_supercolumn_to_array($c_or_sc) {
		$ret = array();
		if ($c_or_sc->column) {
			$ret[$c_or_sc->column->name] = $c_or_sc->column->value;
		} else if ($c_or_sc->super_column) {
			$ret[$c_or_sc->super_column->name] = array();
			foreach ($c_or_sc->super_column->columns as $column)
				$ret[$c_or_sc->super_column->name][$column->name] = $column->value;
		} else {
			var_dump($c_or_sc);
		}
		return $ret;
	}
}